package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: guoml
 * @Description: 针对企查查t_annual_socialsecurity表和中数t_an_socsecinfo表进行数据映射开发
 * @Date: 2021/12/7 12:51
 */

public class TAnnualSocialsecurity {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_annual_socialsecurity").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_an_socsecinfo");
        long zs_count = zs_data.count();
        if(zs_count>0){

            //获取中数增量数据表中的新增数据（left join）
            Dataset<Row> insert_data = spark.sql("SELECT b.ANCHEID,\n" +
                    "       d.key_no,\n" +
                    "       d.company_id,\n" +
                    "       b.entname,\n" +
                    "       b.SO110,\n" +
                    "       b.SO210,\n" +
                    "       b.SO310,\n" +
                    "       b.SO410,\n" +
                    "       b.SO510,\n" +
                    "       b.TOTALWAGES_SO110,\n" +
                    "       b.TOTALWAGES_SO210,\n" +
                    "       b.TOTALWAGES_SO310,\n" +
                    "       b.TOTALWAGES_SO410,\n" +
                    "       b.TOTALWAGES_SO510,\n" +
                    "       b.TOTALPAYMENT_SO110,\n" +
                    "       b.TOTALPAYMENT_SO210,\n" +
                    "       b.TOTALPAYMENT_SO310,\n" +
                    "       b.TOTALPAYMENT_SO410,\n" +
                    "       b.TOTALPAYMENT_SO510,\n" +
                    "       b.UNPAIDSOCIALINS_SO110,\n" +
                    "       b.UNPAIDSOCIALINS_SO210,\n" +
                    "       b.UNPAIDSOCIALINS_SO310,\n" +
                    "       b.UNPAIDSOCIALINS_SO410,\n" +
                    "       b.UNPAIDSOCIALINS_SO510,\n" +
                    "       b.JOBID,\n" +
                    "       \"1\" as isadd\n" +
                    "FROM\n" +
                    "  (SELECT m.*,n.CREDITNO,n.REGNO,n.ENTNAME\n" +
                    "   FROM a_dc_ep_ods.zs_t_an_socsecinfo m\n" +
                    "INNER JOIN\n" +
                    "  a_dc_ep_ods.zs_t_an_basic n on m.ANCHEID = n.ANCHEID\n" +
                    "   WHERE n.CREDITNO IS NOT NULL\n" +
                    "     AND n.CREDITNO !='') b\n" +
                    "LEFT JOIN\n" +
                    "     a_dc_ep_ods.t_eci_company d ON upper(trim(d.credit_code)) = upper(trim(b.CREDITNO))\n" +
                    "     AND d.credit_code IS NOT NULL and d.credit_code!=''").union(spark.sql("SELECT b.ANCHEID,\n" +
                    "       d.key_no,\n" +
                    "       d.company_id,\n" +
                    "       b.entname,\n" +
                    "       b.SO110,\n" +
                    "       b.SO210,\n" +
                    "       b.SO310,\n" +
                    "       b.SO410,\n" +
                    "       b.SO510,\n" +
                    "       b.TOTALWAGES_SO110,\n" +
                    "       b.TOTALWAGES_SO210,\n" +
                    "       b.TOTALWAGES_SO310,\n" +
                    "       b.TOTALWAGES_SO410,\n" +
                    "       b.TOTALWAGES_SO510,\n" +
                    "       b.TOTALPAYMENT_SO110,\n" +
                    "       b.TOTALPAYMENT_SO210,\n" +
                    "       b.TOTALPAYMENT_SO310,\n" +
                    "       b.TOTALPAYMENT_SO410,\n" +
                    "       b.TOTALPAYMENT_SO510,\n" +
                    "       b.UNPAIDSOCIALINS_SO110,\n" +
                    "       b.UNPAIDSOCIALINS_SO210,\n" +
                    "       b.UNPAIDSOCIALINS_SO310,\n" +
                    "       b.UNPAIDSOCIALINS_SO410,\n" +
                    "       b.UNPAIDSOCIALINS_SO510,\n" +
                    "       b.JOBID,\n" +
                    "       \"1\" AS isadd\n" +
                    "FROM\n" +
                    "  (SELECT m.*,n.CREDITNO,n.REGNO,n.ENTNAME\n" +
                    "   FROM a_dc_ep_ods.zs_t_an_socsecinfo m\n" +
                    "INNER JOIN\n" +
                    "  a_dc_ep_ods.zs_t_an_basic n on m.ANCHEID = n.ANCHEID\n" +
                    "   WHERE n.regno IS NOT NULL\n" +
                    "     AND n.regno !=''\n" +
                    "     AND (n.CREDITNO IS NULL\n" +
                    "          OR n.CREDITNO = '')) b\n" +
                    "LEFT JOIN\n" +
                    "     a_dc_ep_ods.t_eci_company d ON upper(trim(d.no)) = upper(trim(b.regno))\n" +
                    "     AND d.no IS NOT NULL and d.no!=''"));
            insert_data.createOrReplaceTempView("tmp_t_annual_socialsecurity1");
            spark.sql("insert overwrite table a_dc_ep_incr.t_annual_socialsecurity select * from tmp_t_annual_socialsecurity1");

        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_annual_socialsecurity");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
